package wsz;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;
import java.util.Map;

/**
 * 单个producer，使用事务保证消息的仅一次发送
 * @author wsz
 * @date 2022/3/18 11:05
 **/
public class MyTransactionProducer {

    public static void main(String[] args) {
        Map<String, Object> configs = new HashMap<>();
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.174.138:9092");
        configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        // 提供客户端ID
        configs.put(ProducerConfig.CLIENT_ID_CONFIG, "tx_producer");
        // 事务ID
        configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "my_tx_id");
        // 要求ISR都确认
        configs.put(ProducerConfig.ACKS_CONFIG, "all");

        KafkaProducer<Object, String> producer = new KafkaProducer<>(configs);
        // 初始化事务
        producer.initTransactions();

        // 开启事务
        producer.beginTransaction();

        try {
            producer.send(new ProducerRecord<>("tp_tx_01", "tx_msg_01"));
//             int i = 1 / 0;
            // 提交事务
            producer.commitTransaction();
        } catch (Exception ex) {
            ex.printStackTrace();
            // 中⽌事务
            producer.abortTransaction();
        } finally {
            // 关闭⽣产者
            producer.close();
        }
    }
}
